"""
Python高级特性学习 - 异步编程
====================

本文件演示了Python中异步编程的核心概念和高级特性，包括：
1. 基本的async/await语法
2. 异步生成器和异步迭代
3. 生产者-消费者模式的异步实现
4. 异步上下文管理器
5. 超时处理和异常管理
6. 并发控制和任务调度

通过这些示例，可以深入理解Python异步编程的工作原理和最佳实践。
"""

import asyncio
import time
from typing import AsyncGenerator, List
from contextlib import asynccontextmanager


# 1. 基本异步函数示例
async def basic_async_function() -> str:
    """基本异步函数示例

    演示async/await的基本用法，模拟异步IO操作。
    """
    print("开始异步操作...")
    await asyncio.sleep(1)  # 模拟异步IO操作
    print("异步操作完成")
    return "异步操作结果"


# 2. 异步生成器示例
async def async_number_generator(count: int) -> AsyncGenerator[int, None]:
    """异步生成器示例

    异步生成一系列数字，每次生成前等待一段时间。

    Args:
        count: 生成数字的数量

    Yields:
        int: 生成的数字
    """
    for i in range(count):
        await asyncio.sleep(0.5)  # 模拟异步IO操作
        yield i
        print(f"生成数字: {i}")


# 3. 生产者-消费者模式
async def producer(queue: asyncio.Queue, num_items: int) -> None:
    """生产者函数

    向队列中添加数据项。

    Args:
        queue: 用于存储数据的异步队列
        num_items: 要生产的项目数量
    """
    for i in range(num_items):
        # 模拟生产数据的时间
        await asyncio.sleep(0.5)
        item = f"item-{i}"
        await queue.put(item)
        print(f"生产者生产了: {item}")

    # 发送结束信号
    await queue.put(None)
    print("生产者完成")


async def consumer(queue: asyncio.Queue, consumer_id: int) -> None:
    """消费者函数

    从队列中获取并处理数据项。

    Args:
        queue: 用于获取数据的异步队列
        consumer_id: 消费者ID
    """
    while True:
        # 从队列中获取数据项
        item = await queue.get()

        # 检查结束信号
        if item is None:
            # 将结束信号放回队列供其他消费者使用
            await queue.put(None)
            print(f"消费者{consumer_id}完成")
            break

        # 模拟处理数据的时间
        await asyncio.sleep(1)
        print(f"消费者{consumer_id}处理了: {item}")

        # 标记任务完成
        queue.task_done()


# 4. 异步上下文管理器
@asynccontextmanager
async def async_timer() -> AsyncGenerator[float, None]:
    """异步计时器上下文管理器

    异步版本的计时器，用于测量异步代码块的执行时间。

    Yields:
        float: 代码块执行时间
    """
    start_time = time.time()
    print("开始计时...")
    try:
        yield start_time
    finally:
        end_time = time.time()
        print(f"执行时间: {end_time - start_time:.2f}秒")


# 5. 超时处理示例
async def long_running_task(duration: int) -> str:
    """长时间运行的任务

    模拟一个需要较长时间才能完成的任务。

    Args:
        duration: 任务持续时间（秒）

    Returns:
        str: 任务完成消息
    """
    print(f"开始执行长时间任务，预计耗时{duration}秒...")
    await asyncio.sleep(duration)
    print("长时间任务完成")
    return "长时间任务完成结果"


async def timeout_demo() -> None:
    """超时处理演示

    演示如何使用asyncio.wait_for处理异步任务的超时。
    """
    print("=== 超时处理演示 ===")

    # 1. 正常完成的任务
    try:
        result = await asyncio.wait_for(long_running_task(2), timeout=5.0)
        print(f"任务正常完成: {result}")
    except asyncio.TimeoutError:
        print("任务超时")

    # 2. 超时的任务
    try:
        result = await asyncio.wait_for(long_running_task(5), timeout=2.0)
        print(f"任务正常完成: {result}")
    except asyncio.TimeoutError:
        print("任务超时")


# 6. 异步编程最佳实践
async def best_practices_demo() -> None:
    """异步编程最佳实践演示

    展示异步编程中的一些重要最佳实践。
    """
    print("\n=== 异步编程最佳实践 ===")

    # 1. 使用asyncio.gather并发执行多个任务
    print("1. 并发执行多个任务:")
    async def task(name: str, duration: float) -> str:
        print(f"任务 {name} 开始")
        await asyncio.sleep(duration)
        print(f"任务 {name} 完成")
        return f"结果_{name}"

    # 并发执行三个任务
    results = await asyncio.gather(
        task("A", 1.0),
        task("B", 1.5),
        task("C", 2.0)
    )
    print(f"所有任务完成，结果: {results}")

    # 2. 使用asyncio.as_completed处理完成的任务
    print("\n2. 处理最先完成的任务:")
    tasks = [
        task("X", 1.0),
        task("Y", 1.5),
        task("Z", 0.5)
    ]

    for coro in asyncio.as_completed(tasks):
        result = await coro
        print(f"任务完成，结果: {result}")

    # 3. 使用asyncio.Semaphore限制并发数量
    print("\n3. 限制并发数量:")
    semaphore = asyncio.Semaphore(2)  # 最多同时运行2个任务

    async def limited_task(name: str) -> str:
        async with semaphore:  # 获取信号量
            print(f"任务 {name} 开始执行")
            await asyncio.sleep(1.0)
            print(f"任务 {name} 执行完成")
            return f"结果_{name}"

    # 创建多个任务
    limited_tasks = [limited_task(f"L{i}") for i in range(5)]
    limited_results = await asyncio.gather(*limited_tasks)
    print(f"限制并发任务完成，结果: {limited_results}")


# 7. 主演示函数
async def main_demo() -> None:
    """主演示函数

    按顺序执行所有异步编程示例。
    """
    print("Python异步编程学习示例\n" + "=" * 30)

    # 1. 基本异步函数
    print("\n1. 基本异步函数:")
    async with async_timer() as start_time:
        result = await basic_async_function()
        print(f"返回结果: {result}")

    # 2. 异步生成器
    print("\n2. 异步生成器:")
    async with async_timer() as start_time:
        async for number in async_number_generator(3):
            print(f"接收到数字: {number}")

    # 3. 生产者-消费者模式
    print("\n3. 生产者-消费者模式:")
    async with async_timer() as start_time:
        # 创建队列
        queue = asyncio.Queue(maxsize=5)

        # 创建生产者和消费者任务
        producer_task = asyncio.create_task(producer(queue, 5))
        consumer_tasks = [
            asyncio.create_task(consumer(queue, 1)),
            asyncio.create_task(consumer(queue, 2))
        ]

        # 等待所有任务完成
        await producer_task
        await asyncio.gather(*consumer_tasks)

    # 4. 超时处理
    await timeout_demo()

    # 5. 最佳实践
    await best_practices_demo()

    print("\n所有异步编程示例执行完成！")


def main() -> None:
    """主函数

    程序入口点，运行所有异步编程示例。
    """
    # 使用asyncio.run()运行主异步函数
    asyncio.run(main_demo())


if __name__ == "__main__":
    main()